package com.tongcao.cn.mq.listner;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.aliyun.openservices.ons.api.*;
import com.tongcao.cn.mq.to.AsyncMsgTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * @Author:tongcao
 * @Date:2019/1/25 17:08
 * Description:ONS
 */
public class ConsumerListener extends BaseConsumerListener {

    private Logger logger = LoggerFactory.getLogger(ConsumerListener.class);
    private Consumer consumer;

    /**
     * Listener启动
     */
    @Override
    void start() throws IOException, MQClientException {
        try {
            consumer = ONSFactory.createConsumer(consumerProperties);
            //订阅消息
            consumer.subscribe(topic, messageHandle.getSubExpression(), new MessageListener() {
                @Override
                public Action consume(Message message, ConsumeContext context) {
                    AsyncMsgTO msgTO = new AsyncMsgTO(message);
                    boolean ret = messageHandle.consume(msgTO);
                    if (logger.isDebugEnabled()) {
                        logger.debug("收到一条消息,msg:" + msgTO.toString() + ",处理结果:" + ret);
                    }
                    return ret ? Action.CommitMessage : Action.ReconsumeLater;
                }
            });
            logger.info("consumer started!topic:" + topic + ",expression:" + messageHandle.getSubExpression()
                    + ",properties:" + consumerProperties.toString());
            consumer.start();
        } catch (Exception e) {
            logger.error("ConsumerListener start error!", e);
        }
    }

    @Override
    public void close() {
        consumer.shutdown();
    }
}
